#include <QUuid>
#include <QTimer>
#include <QTimerEvent>

#include "rabbitmqclient.h"

RabbitMqClient::RabbitMqClient(QObject *parent) : QObject(parent),m_requestId(0),m_start(false)
{

}

void RabbitMqClient::start(QString uri) {
    if(!m_start)
    {
        connect(this, SIGNAL(requestSignal(qint64,QString,int)), this, SLOT(handleRequest(qint64,QString,int)));
        connect(&m_client, SIGNAL(connected()), this, SLOT(clientConnected()));
        m_client.connectToHost(uri);
        m_start = true;
    }
}

void RabbitMqClient::stop() {
    if(m_start)
    {
        m_start = false;
        disconnect(this, SIGNAL(requestSignal(qint64,QString,int)), 0,0);
        disconnect(&m_client, SIGNAL(connected()), 0,0);
        m_requestHash.clear();
        m_requestTimerHash.clear();
        m_client.disconnectFromHost();
        QMutexLocker lockData( &m_mutex);
        m_requestId = 0;
    }
}

qint64 RabbitMqClient::request(QString msg, int timeout)
{
    QMutexLocker lockData( &m_mutex);
    m_requestId++;
    emit requestSignal(m_requestId, msg,timeout);
    return m_requestId;
}


void RabbitMqClient::handleRequest(qint64 requestId, QString msg,int timeout)
{
    QString routingKey = m_servicename + ".request";
    QString correlationId = QUuid::createUuid().toString();
    m_requestHash.insert(correlationId,requestId);
    QAmqpMessage::PropertyHash properties;
    properties.insert(QAmqpMessage::ReplyTo, m_responsecQueue->name());
    properties.insert(QAmqpMessage::CorrelationId, correlationId);
    m_cmdExchange->publish(msg, routingKey, properties);
    int timerId = startTimer(timeout);
    m_requestTimerHash.insert(timerId,correlationId);
}

void RabbitMqClient::clientConnected() {
    QString exchangeName = m_servicename + ".topic";
    QAmqpExchange *topic_exchange = m_client.createExchange(exchangeName);
    connect(topic_exchange, SIGNAL(declared()), this, SLOT(topicExchangeDeclared()));
    topic_exchange->declare(QAmqpExchange::Topic);

    exchangeName = m_servicename + ".cmd";
    m_cmdExchange = m_client.createExchange(exchangeName);
    connect(m_cmdExchange, SIGNAL(declared()), this, SLOT(cmdExchangeDeclared()));
    m_cmdExchange->declare(QAmqpExchange::Direct);

}

void RabbitMqClient::topicExchangeDeclared() {
    QAmqpQueue *temporaryQueue = m_client.createQueue();
    connect(temporaryQueue, SIGNAL(declared()), this, SLOT(topicQueueDeclared()));
    connect(temporaryQueue, SIGNAL(bound()), this, SLOT(topicQueueBound()));
    connect(temporaryQueue, SIGNAL(messageReceived()), this, SLOT(topicMessageReceived()));
    temporaryQueue->declare(QAmqpQueue::Exclusive);
}

void RabbitMqClient::cmdExchangeDeclared() {
    QAmqpQueue *temporaryQueue = m_client.createQueue();
    connect(temporaryQueue, SIGNAL(declared()), this, SLOT(cmdQueueDeclared()));
    connect(temporaryQueue, SIGNAL(bound()), this, SLOT(cmdQueueBound()));
    connect(temporaryQueue, SIGNAL(messageReceived()), this, SLOT(responseReceived()));
    temporaryQueue->declare(QAmqpQueue::Exclusive);
}


void RabbitMqClient::topicQueueDeclared() {
    QAmqpQueue *temporaryQueue = qobject_cast<QAmqpQueue*>(sender());
    if (!temporaryQueue)
        return;
    QString exchangeName = m_servicename + ".topic";
    QString keystr = m_servicename;
    temporaryQueue->bind(exchangeName, keystr);
    qDebug() << exchangeName << " QueueDeclared";
}

void RabbitMqClient::cmdQueueDeclared() {
    QAmqpQueue *temporaryQueue = qobject_cast<QAmqpQueue*>(sender());
    if (!temporaryQueue)
        return;
    m_responsecQueue = temporaryQueue;
    temporaryQueue->bind(m_cmdExchange->name(), temporaryQueue->name());
    qDebug() << m_cmdExchange->name() << " QueueDeclared";
}

void RabbitMqClient::topicQueueBound() {
    QAmqpQueue *temporaryQueue = qobject_cast<QAmqpQueue*>(sender());
    if (!temporaryQueue)
        return;
    temporaryQueue->consume(QAmqpQueue::coNoAck);
}

void RabbitMqClient::cmdQueueBound() {
    QAmqpQueue *temporaryQueue = qobject_cast<QAmqpQueue*>(sender());
    if (!temporaryQueue)
        return;
    temporaryQueue->consume(QAmqpQueue::coNoAck);
}

void RabbitMqClient::topicMessageReceived() {
    QAmqpQueue *temporaryQueue = qobject_cast<QAmqpQueue*>(sender());
    if (!temporaryQueue)
        return;

    QAmqpMessage message = temporaryQueue->dequeue();
    emit subscribedMsg(message.payload());
}

void RabbitMqClient::responseReceived() {
    QAmqpQueue *temporaryQueue = qobject_cast<QAmqpQueue*>(sender());
    if (!temporaryQueue)
        return;
    QAmqpMessage message = temporaryQueue->dequeue();
    QString correlationId = message.property(QAmqpMessage::CorrelationId).toString();
    qDebug() << " responseReceived:[x] routingKey:" << message.routingKey() << ":"  << "correlationId:" << correlationId << " msg:" << message.payload();
    if(m_requestHash.contains(correlationId))
    {
        qint64 reqId = m_requestHash[correlationId];
        QString resp = message.payload();
        emit responseMsg(reqId,resp);
        m_requestHash.remove(correlationId);
    }
    else
    {
        temporaryQueue->reject(message, true);
        qWarning() << " responseReceived:[reject]: ";
        return;
    }
}
void RabbitMqClient::timerEvent(QTimerEvent *event)
{
    int tid = event->timerId();
    if(m_requestTimerHash.contains(tid))
    {
        QString correlationId = m_requestTimerHash[tid];
        m_requestTimerHash.remove(tid);
        if(m_requestHash.contains(correlationId))
        {
            qint64 reqId = m_requestHash[correlationId];
            m_requestHash.remove(correlationId);
            emit responseTimeout(reqId);
            qWarning() << "request timeout id:" << reqId;
        }
    }

}

QString RabbitMqClient::getServiceName() const
{
    return m_servicename;
}

void RabbitMqClient::setServiceName(const QString& newServicename)
{
    m_servicename = newServicename;
}
